package com.reactor.operator;

import java.time.Duration;
import java.time.temporal.ChronoUnit;

import org.junit.Test;

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

/**flatMap通常用于每个元素又会引入数据流的情况，比如我们有一串url数据流，需要请求每个url并收集response数据。假设响应式的请求方法如下：

Mono<HttpResponse> requestUrl(String url) {...}
而url数据流为一个Flux<String> urlFlux，那么为了得到所有的HttpResponse，就需要用到flatMap：

urlFlux.flatMap(url -> requestUrl(url));
其返回内容为Flux<HttpResponse>类型的HttpResponse流。
 * @author DELL
 *
 */
public class FlatMap {

	public static void main(String[] args) {
		//flatMap 和 flatMapSequential 操作符把流中的每个元素转换成一个流，再把所有流中的元素进行合并。flatMapSequential 和 flatMap 之间的区别与 mergeSequential 和 merge 之间的区别是一样的。
		//在代码清单 12 中，流中的元素被转换成每隔 100 毫秒产生的数量不同的流，再进行合并。由于第一个流中包含的元素数量较少，所以在结果流中一开始是两个流的元素交织在一起，然后就只有第二个流中的元素。
		//FlatMap将一个发送事件的上游Observable变换为多个发送事件的Observables，然后将它们发射的事件合并后放进一个单独的Observable里.
		//这句话比较难以理解, 我们先通俗易懂的图片来详细的讲解一下, 首先先来看看整体的一个图片:
		//先看看上游, 上游发送了三个事件, 分别是1,2,3, 注意它们的颜色。中间flatMap的作用是将圆形的事件转换为一个发送矩形事件和三角形事件的新的上游Observable.
		//上游每发送一个事件, flatMap都将创建一个新的水管, 然后发送转换之后的新的事件, 下游接收到的就是这些新的水管发送的数据. 这里需要注意的是, flatMap并不保证事件的顺序, 也就是图中所看到的, 并不是事件1就在事件2的前面.
		Flux.just(5, 10)
	        .flatMap(x -> Flux.interval(Duration.of(x * 10, ChronoUnit.MILLIS), Duration.of(100, ChronoUnit.MILLIS)).take(x))
	        .toStream()
	        .forEach(System.out::println);
	}

	@Test
	// 打印结果为mfolnuox，原因在于各个拆分后的小字符串都是间隔100ms发出的，因此会交叉。
	public void test(){
		StepVerifier.create(
		        Flux.just("flux", "mono")
		                .flatMap(s -> Flux.fromArray(s.split("\\s*"))   // 对于每一个字符串s，将其拆分为包含一个字符的字符串流
		                        .delayElements(Duration.ofMillis(100))) // 对每个元素延迟100ms
		                .doOnNext(System.out::print)) // 对每个元素进行打印（注doOnNext方法是“偷窥式”的方法，不会消费数据流）
		        .expectNextCount(8) // 验证是否发出了8个元素
		        .verifyComplete();
	}

}
